Frameworks

In the Mesos world, we assume that our distributed systems have a central controller and many workers. This is how most successful, scalable systems work: Google’s BigTable, Apache Hadoop, and Twitter’s Storm all have some sort of central controller that organizes their workers. Of course, the workers should be designed to work independently from the controller so that they’re not bottlenecked by it and don’t rely on it for their availability. Applications that run on Mesos are called frameworks. A framework has two parts: the controller portion, which is called the scheduler, and the worker portion, which are called the executors.

To run a framework on a Mesos cluster, we must run its scheduler. A scheduler is simply a process that can speak the Mesos protocol; frequently, our schedulers run on Marathon, but it’s always necessary to consider how to ensure high availability for the scheduler. When a scheduler first starts up, it connects to the Mesos cluster so that it can use the cluster’s resources. As the scheduler runs, it makes requests to Mesos to launch executors as it sees fit. This complete flexibility for the scheduler is what makes Mesos so powerful: schedulers can launch tasks based on resource availability, changing workloads, or external triggers. Today, there exist schedulers that can manage fleets of web services, schedulers that coordinate Storm topologies, and schedulers that optimize batch job placement.

When a scheduler wants to do some work, it launches an executor. The executor is simply the scheduler’s worker: the scheduler then decides to send one or more tasks to the executor, which will work on those tasks independently, sending status updates to the scheduler until the tasks are complete. We’ll learn more about the distinction between tasks and executors in “Tasks and Executors”.

Now that we’ve got an idea of what a framework is, let’s take a step down the stack to learn about the Mesos cluster itself.

Masters and Slaves

A Mesos cluster is comprised of two components: the Mesos masters and the Mesos slaves. The masters are the software that coordinates the cluster; the slaves are what execute code in containers. You might be thinking, “That sounds awfully similar to how frameworks work.”

In that case, you’d be 100% correct. The Mesos cluster mirrors the organization of the frameworks that run on top of it, because the master/slave pattern is simple, flexible, and effective. The cluster software needs to be separate, however, to ensure that frameworks remain isolated and secure from one another. First, we’ll look at a few key ideas about the Mesos masters; then, we’ll take a look at the slaves.

Resources

The fundamental abstraction in Mesos is that of the resource—tasks and executors consume resources while performing their work. A common question is: “What is a resource?” The answer is broad: a resource is anything that a task or executor uses while doing its work. For example, the standard resources (and thus the resources that nearly every framework uses) are cpus, mem (memory), disk, and ports. The first three of those resources are scalar: they represent a total amount available, from which tasks deduct their usage. There are other resource types available in Mesos, as well. Ranges are lists of integers from which single elements or subranges can be allocated; for example, these are used to represent the ports available and in use. The last resource type, sets, are simply unordered collections of strings. We’ll see an example of a set later on. First, let’s learn about the standard resources:

cpus

This resource expresses how many CPU cores are available. Tasks may use fractional parts of a CPU—this is possible because Mesos slaves use CPU shares, rather than reserving specific CPUs. This means that, if you have 1.5 cpus reserved, your processes will be allowed to use a total of 1.5 seconds of CPU time each second. That could mean that, within a single executor, two processes each get 750 milliseconds of CPU time per second, or one process gets 1 second of CPU time and another gets 500 milliseconds of CPU time each in a given second. The benefit of using CPU shares is that if some task would be able to utilize more than its share, and no other task would use an otherwise idle CPU, the first task can potentially use more than its share. As a result, the cpus reserved provides a guaranteed minimum of CPU time available to the task—if additional capacity is available, it will be allowed to use more.

mem

mem is how many megabytes of memory are available. Unlike CPUs, which can burst to use more than their allocation if there’s spare capacity, memory is a strictly preallocated resource. As a result, choosing the correct memory for tasks can be tricky: too little, and the tasks get killed when they try to allocate more than their allotment; too much, and memory sits idle that could otherwise be used for running more tasks.

disk

disk represents the amount of scratch space (in megabytes) that should be available in the container’s sandbox. Since disk usage can be highly variable due to the unpredictability of errors filling up logs, many framework developers don’t worry about reserving this resource in their tasks. Furthermore, by default, Mesos slaves don’t even enforce when a task uses more disk than allocated. (To change that behavior, you can pass the option --enforce_container_disk_quota on your slaves’ command lines. The other reason that this value is ignored is that the Mesos slaves automatically garbage collect the sandboxes of finished tasks when they need to reclaim disk space (see the following sidebar for details and advice on configuring garbage collection for slaves).

ports

Ports are the first (and only default) nonscalar resource on Mesos slaves that we’ll need to understand. Many tasks, such as HTTP servers, will want to open one or more ports, so that they can listen for incoming connections. Since each port can only be bound to one task, Mesos helps by allowing tasks to reserve specific ports, so that each task uses a nonconflicting range. Furthermore, if a task must be bound to a particular port, then it can decline offers that don’t include that port in the available range. By default, Mesos slaves expose ports 31000–32000; however, you can change this range by setting the ports resource to whatever you’d like on the command line.

Now that we’ve seen the basics of Mesos slave resources and learned about their default behaviors, let’s learn how we can customize the resources offered.

--resources='cpus:4;mem:8192;ports:[1000-5000;31000-32000];gpgpus:{gpu1,gpu2}'

--attributes='operating_system:ubuntu;cpu_class:haswell;zone:us_east;rack:22'

Roles

In order to decide which resources can be offered to which frameworks, Mesos includes the concept of “roles.” A role is like a group: you can have the dgrnbrg role (for things you run), the qa role (for quality assurance-related tasks), the db role (for databases), and so on. Besides being used to exclusively reserve resources, roles also can be used to assign weights to Mesos’s fair sharing algorithm. This allows you to specify that some roles should receive a greater ratio of resources on the cluster than others. When registering a framework, you may specify which role it belongs to by setting the role field in the FrameworkInfo message used to register. Frameworks will only receive offers that are for their role, or for the * role.

If you’d like to use roles, then you must include a comma-separated list of the valid roles on the master. For example, if you’d like to have dev, qa, and prod roles, then you’d need to pass --roles=dev,qa,prod to the master. To assign weights to those roles, so that dev gets twice as many offers as qa and prod gets three times as many offers as qa, you’d pass --weights='dev=2,qa=1,prod=3'.

--resources='cpus(prod):1;mem(prod):4096; \
  cpus(qa):2;mem(qa):1024;cpus:13;mem:11264'

--default_role=prod \
  --resources='cpus:1;mem:4096;cpus(qa):2;mem(qa):1024;cpus(*):13;mem(*):11264'

[
    {
        "name": "cpus",
        "type": "SCALAR", 
        "scalar": { "value": 1 },
        "role": "prod",
        "reservation": {
            "principal": "admin" 
        }
    },
    {
        "name": "mem",
        "type": "SCALAR",
        "scalar": { "value": 4096 },
        "role": "prod",
        "reservation": {
            "principal": "admin"
        }
    },
    {
        "name": "cpus",
        "type": "SCALAR",
        "scalar": { "value": 2 },
        "role": "qa",
        "reservation": {
            "principal": "admin"
        }
    },
    {
        "name": "mem",
        "type": "SCALAR",
        "scalar": { "value": 1024 },
        "role": "qa",
        "reservation": {
            "principal": "admin"
        }
    }

curl -u admin:password \ 
     -d slaveId=$SLAVEID\ 
     -d resources="$(cat resources.json)"\ 
     -X POST\ 
     http://<ip>:<port>/master/reserve 

curl -u admin:password \
     -d slaveId=$SLAVEID\
     -d resources="$(cat resources.json)"\
     -X POST\
     http://<ip>:<port>/master/unreserve 

Tasks and Executors

When a scheduler wants to do some work, it launches an executor. The executor is simply the scheduler’s worker: the scheduler can decide to send one or more tasks to an executor, which will work on those tasks independently, sending status updates to the scheduler until the tasks are complete.

Up to this point, we haven’t tried to distinguish the ideas of tasks and executors, because usually each executor has a single task, so there’s not much point in differentiating between them. However, some frameworks leverage the full power of the task and executor abstraction, and to understand these frameworks we must understand the abstraction. For that we must first understand the problem: tasks are things that our scheduler wants to run in a container on a slave; however, we’d sometimes like to be able to run multiple tasks in the same container. For example, this pattern shows up with Storm workers and Hadoop TaskTrackers, each of which can contain many tasks.

So, what’s an executor? An executor is a process container that runs tasks.

And what’s a task? A task is the unit of work in Mesos.

Figure 2-1 illustrates the relationship between the elements of this abstraction.

Figure 2-1. Relationship between slaves (machines), executors (containers), and tasks

There are also some caveats that you should be aware of:

Tasks can adjust the size of the executor’s container

When you launch a task or executor in Mesos, both have resources attached to them. The reason for this is that an executor has a basic amount of resources that it requires for itself, and then each task can use additional resources. Mesos will automatically expand and contract the executor’s container to have resources equal to the sum of those required by the executor plus all of its running tasks.

APIs can be task or executor oriented

When programming for the executor API (see Chapter 5 for details), some communication APIs between the executor and scheduler are are specified per task (such as launchTask and statusUpdate), whereas other APIs are per executor (such as frameworkMessage). It’s up to the framework developers to make sure that they associate messages with tasks whenever necessary, regardless of whether the Mesos API does that automatically.

Most frameworks don’t worry about distinguishing tasks and executors

Nearly every Mesos framework has one task for each executor. This means they don’t need to worry about various idiosyncrasies in the executor APIs. Unfortunately, that in turn means that occasionally the executor APIs can be difficult to understand, since they see less use.

Understanding mesos.proto

Throughout this book and your own development efforts, you’ll often wonder how you configure some API or what the configuration options are for some feature. All communication with Mesos is encoded in Google Protocol Buffers (commonly known as protobufs). Lucky for us, all of the data schemas are stored in a central location: the file include/mesos/mesos.proto in the Mesos codebase. Let’s take a tour of Google protobufs, and learn some idioms that frequently show up in the Mesos codebase.

Protobufs is an interface description language that allows an application’s developers to specify strongly typed messages that exist in their system. Many languages have protobuf compilers, which convert .proto files into efficient serializers and deserializers so that all of the languages can communicate with an efficient binary encoding.

First, let’s learn the basics of protobufs:

/** 
 *  This represents information about a user.
 *  We'll respect their privacy as much as possible.
 */
message UserInfo { 
  // The name of the user 
  required string username = 1; 
  required Name name = 2; 
  optional uint32 age = 4; 
  // Auxiliary info
  optional bytes aux_info = 3; 
}

message Name { 
  optional string first = 1; 
  optional string last = 2;

  message MiddleName { 
    optional string regular = 1; 
    optional string patronymic = 2;
  }

  repeated MiddleName middle = 3; 

  optional uint32 total_parts = 4 [default = 2]; 
}

Protobufs supports block comments in the style of JavaDocs.

The main syntactic form in a .proto file is the message, which has a name (UserInfo) and a set of typed, named fields.

Protobufs also supports comments that span to the end of the line, like C and Java.

A field has four parts: whether is it required, optional, or repeated; the type (string, in this case); the name of the field; and the field index. Each field index must be unique to the message; it is the developer’s responsibility to never reuse the indices, because they are the foundation on which backward compatibility in the binary format is built.

A field’s type can itself be another message.

Protobufs has many built-in numeric types; Mesos primarily uses int32, uint32, and double. Also, notice that the field indices don’t have to be in numeric order—they’re usually grouped to reflect the structure of the message.

bytes allows you to store arbitrary binary data in a protobuf. In Mesos’s protobuf messages, this is used to enable developers to define custom additional payloads.

Generally, most or all of a project’s messages are defined in a single file.

You don’t need to make any fields required.

Messages can be nested into other messages.

Note that the field numbering restarts within every message, even nested messages.

Fields can also be repeated, in which case they’re a list of zero or more entries of the specified type (in this case, MiddleName).

Protobufs allows default values to be specified.

Besides the basics, there are a few patterns that frequently show up in mesos.proto. Firstly, any ID in the system typically has a message defined for it, with only a single required string value (c.f. FrameworkID, TaskID). This is beneficial for typed languages like C++, Scala, and Java, because it ensures that different types of IDs can’t get mixed up. Secondly, when there is a choice for what to specify (for instance, a custom executor vs. a command executor), both choices are made optional, and there should be a comment in mesos.proto that explains the choice. Most messages that can be generated by framework code have a data field of type bytes: this is the place to put any serialized messages or data you may want to transmit.

Another common pattern in mesos.proto is tagged unions. A tagged union is also known as a discriminated union; it is used when there are multiple mutually exclusive choices for how to configure something. Unions of this type, such as the Resource message, have every choice optional, so that you can fill in just one per message. There is also a required Type field, which indicates to the code which field you intended to specify. For example, to specify a scalar resource, the Type of the resource should be SCALAR and the scalar field should contain the value.

Finally, mesos.proto is usually where new features of Mesos are specified and incubated. As a result, you’ll often see upcoming features described in mesos.proto without other official documentation. For example, the service discovery system and the persistent volumes features were in mesos.proto for several releases prior to becoming stable, supported features in Mesos. mesos.proto can give you a lot of insight into what’s coming up for Mesos; however, it can be confusing and overwhelming if you don’t ignore messages and fields unrelated to your task at hand.

Not Managed by Mesos

To have a functional Mesos cluster, you’ll need a few other applications that are difficult or impossible to run as frameworks:

ZooKeeper

Mesos uses ZooKeeper to provide discovery services across your cluster. ZooKeeper quorums are named by the DNS names of the machines that they run on; as a result, the Mesos cluster’s ZooKeeper quorum usually runs on three or five specific machines, so that it can have a stable name.

Databases

Initial support for persistent volumes was only recently added, in Mesos version 0.23. As a result, most databases don’t yet have mature Mesos frameworks. Nevertheless, work is rapidly progressing on hosting databases on Mesos. Twitter has been developing Apache Cotton (previously called Mysos), which runs and manages MySQL on Mesos, since 2014. Mesosphere has also been working with its partners to productionalize Cassandra and Kafka on Mesos, and Basho has released a beta version of a Riak framework for Mesos, which includes a high-performance proxy layer that was enabled by Mesos.

Monitoring

Mesos watches and monitors its frameworks, which helps developers build highly available and reliable systems. But, in the loosely translated words of the Roman poet Juvenal, “Who watches the watchers?” Mesos itself needs to be monitored; it collects instantaneous usage and utilization data, but that data should be archived and graphed to understand the cluster’s behavior over time. In addition, Mesos isn’t built to identify issues with the machines that it’s running on. Thus, a robust, productionalized Mesos cluster should have host-level monitoring, and it should archive and alert on data collected by Mesos itself. One integrated solution to this is Satellite, which can automatically disable flaky slaves, generate alerts on custom conditions, and store historical data to InfluxDB, ElasticSearch, and many other databases.

Luckily, setting up a monitoring system and one ZooKeeper cluster involves a lot less work than setting up standalone versions of all the frameworks that we’re going to run on Mesos!

Summary

A Mesos cluster is comprised of the masters, which coordinate the cluster, and the slaves, which execute tasks on behalf of frameworks. The master/slave structure of the Mesos cluster is mirrored by the frameworks, which are each composed of a scheduler that controls executors, which run on the slaves.

Mesos slaves can be customized with additional key/value metadata, custom resource types, and roles, which are all ways to ensure that framework schedulers can choose the best slaves on which to run their executors, and to help the administrators to guarantee the allocation behaviors of the Mesos masters. The master and slave states can be browsed through web UIs that cross-reference data across the cluster.

Executors are the processes that actually run inside of containers on the slaves. In Mesos, executors can host one or more tasks, which provides flexibility for framework authors to structure their applications however they’d like; however, most frameworks use the CommandExecutor to run one task per executor (i.e., one task per container).

Finally, although Mesos provides a lot of functionality, one still must deploy certain tools outside of Mesos: specifically, a ZooKeeper quorum to bootstrap the cluster, a monitoring system, and possibly some databases.

Now that we’ve had a tour of Mesos, let’s learn how to port an existing web application onto Mesos!

Moving a Web Application to Mesos

Nearly every company has a web application. However, no matter whether it’s written in Ruby on Rails, Scala Play!, or Python’s Flask, deploying it in a reliable, scalable, highly available manner is always a challenge. In this section, we’re going to learn about using Marathon—an easy-to-use Platform as a Service (PaaS) developed by Mesosphere—and how to integrate with other tools, such as HAProxy. Through this process, it is our goal to realize a few benefits over previous architectures:

1. All of our backend processes will be able to run on any machine, and our Mesos framework will handle automatically spinning up new instances of the backend when existing instances fail.

2. We will host our static assets, fast-responding endpoints, dangerous endpoints,¹ and API in entirely different containers to improve our isolation.

3. We will make it easy to deploy a new version or to roll back, and to do so in a matter of seconds.

4. We will put the pieces in place to allow our applications to autoscale based on demand.

At its core, Marathon allows you to specify a command line or Docker image, number of CPUs, amount of memory, and number of instances, and it will start that many instances of that command line or image in containers with the requested amount of CPU and memory. We will explore this basic configuration as well as a few ways to integrate Marathon with load balancers, in order to expose our services internally and to the outside world.

Setting Up Marathon

Obviously, in order to use Marathon, you’ll need to download it. Marathon comes prepackaged for Ubuntu, Debian, Red Hat, and CentOS on the Mesosphere website. Of course, you can also build it yourself, by downloading the latest tarball from Mesosphere’s website, or by cloning it from GitHub. Marathon is written in Scala, so you simply need to run sbt assembly to build it. From here on out, we’ll assume that you’ve got a copy of Marathon installed.

Marathon is trivial to run as a highly available application. In fact, all you need to do to run Marathon in a highly available configuration is to ensure that you start two or three instances of Marathon, and that they all share the same --zk command-line argument. You should also make other options the same, such as the --master, --framework-name, and the ports that it listens for HTTP requests on, or you could see confusing behavior. When running Marathon in this mode, all the instances of Marathon will correctly respond to requests, and they’ll magically synchronize their state so that you don’t need to worry about this. Typically, you’ll finish the setup by assigning the instances of Marathon to a round-robin DNS name, or putting it behind a load balancer. Please see Table 3-1 for further information on selected command-line options.

Using Marathon

Marathon is controlled entirely through its HTTP API. With a working instance of Marathon, let’s try launching an application.

We’ll run a simple application: the famous Python SimpleHTTPServer. This application simply serves the files in the directory in which it was started. We’re going to base our example on it because nearly everyone’s got Python installed, and it’s available by default.

If you run this command:

python -m SimpleHTTPServer 8000

on your computer, then you should be able to navigate to localhost:8000 in your browser and see a directory listing of the folder in which you started the server.

Let’s see how we can run that on a Marathon server running on marathon.example.com:8080. We’ll first make a file containing the JSON descriptor of the Marathon application shown in Example 3-1.

{
    "cmd": "python -m SimpleHTTPServer 31500", 
    "cpus": 0.5, 
    "mem": 50, 
    "instances": 1, 
    "id": "my-first-app", 
    "ports": [31500], 
    "requirePorts": true 
}

The "cmd" specifies the command line that will launch your application.

"cpus" specifies the number of CPUs that your application’s container should have. This number can be fractional.

"mem" specifies the number of MB of memory that your application’s container should have.

"instances" specifies how many copies of this application should be started on the cluster. For this demo, we’ll only make one instance, but typical applications will have anywhere from 2 to 2,000 instances.

"id" is how we’ll refer to this application in the future via the APIs. This should be unique for each application.

"ports" is an array of the ports that the application requires. In this case, we only need a single port. Later, we’ll see how to dynamically allocate and bind to ports.

Since we’re explicitly specifying which ports we want to allocate, we must tell Marathon, or else it will default to automatic port assignment.

Now let’s suppose that data has been stored in a file called my-first-app.json. In order to start this application, we’ll use an HTTP POST to send the data to a Marathon server running at marathon.example.com:8080:

curl -X POST -H 'Content-Type: application/json' \
    marathon.example.com:8080/v2/apps --data @my-first-app.json

This should give us a response that looks something like:

{
   // This section contains parameters we specified
   "id" : "/my-first-app",
   "instances" : 1,
   "cmd" : "python -m SimpleHTTPServer 31500",
   "ports" : [ 31500 ],
   "requirePorts": true,
   "mem" : 50,
   "cpus" : 0.5,

   // This section contains parameters with their automatically assigned values
   "backoffFactor" : 1.15,
   "maxLaunchDelaySeconds" : 3600,
   "upgradeStrategy" : {
      "minimumHealthCapacity" : 1,
      "maximumOverCapacity" : 1
   },
   "version" : "2015-06-04T18:26:18.834Z",
   "deployments" : [
      {
         "id" : "54e5fdf8-8a81-4f95-805f-b9ecc9293095"
      }
   ],
   "backoffSeconds" : 1,
   "disk" : 0,
   "tasksRunning" : 0,
   "tasksHealthy" : 0,
   "tasks" : [],
   "tasksStaged" : 0,

   // This section contains unspecified parameters
   "executor" : "",
   "storeUrls" : [],
   "dependencies" : [],
   "args" : null,
   "healthChecks" : [],
   "uris" : [],
   "env" : {},
   "tasksUnhealthy" : 0,
   "user" : null,
   "requirePorts" : true,
   "container" : null,
   "constraints" : [],
   "labels" : {},
}

which indicates that the application was started correctly.

We can find all the information about the running application by querying for information about my-first-app (what we decided to name this application):

curl marathon.example.com:8080/v2/apps/my-first-app | python -m json.tool

This query returns almost the exact same data as we got when we created the application. The difference is that by now our application has had time to start running, so we can see information about the running tasks:

// ... snip (same as before) ...
    "tasks": [
        {
            "appId": "/my-first-app",
            "host": "10.141.141.10", 
            "id": "my-first-app.7345b7b5-0ae7-11e5-b3a7-56847afe9799", 
            "ports": [ 31500 ], 
            "stagedAt": "2015-06-04T18:28:09.235Z", 
            "startedAt": "2015-06-04T18:28:09.404Z", 
            "version": "2015-06-04T18:28:05.214Z" 
        }
    ],
    "tasksHealthy" : 1, 
// ... snip (same as before) ...

This tells us the DNS name or IP address of the host that is running the task.

The id of the task is a Mesos construct. We can use the Mesos master’s UI to inspect this task; for instance, we can examine its stdout and stderr.

Whether we chose the port that the application is running on or allowed Marathon to choose for us (see Example 3-2 for details), we can find the actually reserved ports for the task.

The staging time of the task is when we submitted the task to a Mesos offer.

The start time of the task is when Mesos actually got around to launching the task.

Every time we update the Marathon descriptor, it gets stamped with a new version (see the response to the initial POST given earlier). Every task is annotated with the version from which it was launched.

This gives a summary of the number of currently healthy tasks in the application.

Let’s look at some of the rich information available about the application. First, we can see all of the configuration parameters that we specified when we created the application. We also see tons and tons of other settings, many of which we’ll go over soon. The next very interesting field is "tasks", which is an array of all of the actual running instances of the application. Each task is represented as a JSON object, which has several useful fields:

• "host" and "ports" tell us what the actual host and ports that were assigned to this task are. These fields are the key to discovering where an application is actually running so that we can connect to it.

• "id" is the mesos TaskID. This is constructed by adding a unique per-task UUID to the application’s ID, which allows for convenient discovery with the Mesos CLI tools or in the Mesos UI.

• "stagedAt" and "startedAt" give us information about when lifecycle events happened to this task. A task is staged once Marathon actually requests Mesos to start the task on a particular offer; a task is started once it begins running.

Usually, you won’t want to hardcode the port for your application: after all, if you did this, then you’d only ever be able to run one instance of the application per slave. Marathon can specify the port for you, instead: simply choose each port your application requires as 0, and the assigned ports will be available in the environment variables $PORT0, $PORT1, etc. If we changed our earlier configuration to take advantage of this, we’d end up with Example 3-2.

{
    "cmd": "./python -m SimpleHTTPServer $PORT0", 
    "cpus": 0.5,
    "mem": 50,
    "instances": 1,
    "id": "my-first-app",
    "ports": [0], 
    "uris": [
        "http://fetchable/uri/with/python.tar.gz" 
    ]
}

Note that we use $PORT0 to access the environment variable that contains the port assigned to the first index of the "ports" array.

We provided 0 as the desired port, meaning that we’re allowing Marathon to choose for us.

We can also provide a list of URIs to be fetched and unzipped into the container before running the command. See “Configuring the process’s environment” for details on the URI schemes supported.

Some applications you’d like to run on Marathon don’t accept their ports via the command line. Don’t worry! There’s an easy way for us to propagate our $PORT0 environment variables into configuration files. I’ll assume that the application’s configuration file is set as a URL to be downloaded with the application. Now, we’re going to use sed to replace a special string in the configuration file with the desired environment variable, be it a $PORT0 or the $MESOS_DIRECTORY:

sed -e 's#@MESOSDIR@#'"$MESOS_DIRECTORY"'#' config.templ > config 
sed -e 's#@PORT@#'"$PORT0"'#' config2.templ > config2 

In our first example, we are replacing the string @MESOSDIR@ with the task’s sandbox directory. We surround the string in the configuration file with @ symbols, because they so rarely show up in configuration languages (of course, you could use any unique string). Also, we use single quotes around the “constants” in our sed command and double quotes around the environment variable to ensure that our shell doesn’t corrupt our command. Finally, we use # instead of / for the sed command so that we can successfully template an environment variable that itself contains forward slashes.

Here, we template the port at index 0 into the configuration file. In fact, we could pass multiple -e arguments to a single invocation of sed if we wanted to template several variables in one configuration file.

We’ll look at other fields as we cover the features that utilize them.

curl -X PUT -H 'Content-Type: application/json' \
    marathon.example.com:8080/v2/apps --data '{"instances": 5}'

curl -X PUT -H 'Content-Type: application/json' \
    marathon.example.com:8080/v2/apps --data '{"instances": 1}'

{
    // ... snip ...
    "constraints": [["dmz", "UNLIKE", "true"],
                    ["hostname", "GROUP_BY"]]
}

{
    // ... snip ...
    "container": {
        "type": "DOCKER", 
        "docker": {
            "image": "group/image", 
            "network": "HOST" 
        }
    }
}

{
    "container": {
        "type": "DOCKER",
        "docker": {
            "image": "group/image",
            "network": "HOST"
        },
        "volumes": [ 
            { 
                "containerPath": "/var/hostlib",
                "hostPath": "/usr/lib",
                "mode": "RO" 
            },
            { 
                "containerPath": "/var/scratch",
                "hostPath": "/mount/ssd",
                "mode": "RW"
            }
        ]
    }
}

{
    "gracePeriodSeconds": 300, 
    "intervalSeconds": 60, 
    "maxConsecutiveFailures": 3, 
    "timeoutSeconds": 30 
}

{
    "protocol": "HTTP", 
    "path": "/healthcheck", 
    "portIndex": 0 
}

{
    "protocol": "TCP", 
    "portIndex": 0 
}

{
    "upgradeStrategy": {
        "minimumHealthCapacity": 0.5,
        "maximumOverCapacity": 0.1
    }
}

Running Mesos Frameworks on Marathon

A very common choice in Mesos clusters is to run the cluster’s Mesos frameworks on Marathon. But Marathon is a Mesos framework itself! So what does it mean to run a Mesos framework on Marathon? Rather than worrying about deploying each framework’s scheduler to specific hosts and dealing with those hosts’ failures, Marathon will ensure that the framework’s scheduler is always running somewhere on the cluster. This greatly simplifies deploying new frameworks in a highly reliable configuration.

{
    "cmd": "java -jar chronos.jar -Xmx350m -Xms350m ---port $PORT0", 
    "uris": [
        "http://fetchable/uri/with/chronos.jar" 
    ],
    "cpus": 1, 
    "mem": 384, 
    "instances": 2, 
    "id": "chronos",
    "ports": [0], 
    "constraints": [["hostname", "UNIQUE"]] 
}

Alternatives to Marathon + Chronos

Of course, Marathon (and Chronos) won’t fit every cluster’s needs. Let’s briefly look at two other popular Mesos PaaS frameworks: Aurora and Singularity. Both frameworks have a much steeper learning curve, but they offer many additional features.

Summary

In this chapter, we learned how to build standard stateless server-based applications with Mesos. What we achieved was self-healing, self-restarting, and simplified service discovery—we basically built a PaaS on Mesos! This was accomplished by using Marathon, a popular Mesos framework.

Marathon launches, monitors, and restarts processes across your Mesos cluster. It simplifies rollout, allows for push-button scale up and scale down, and provides a simple JSON API that’s easy to integrate with.

First, we learned how to start up Marathon, how to secure it, and how to ensure that it’s highly available. Then, we learned how to write a Marathon service descriptor, which allowed us to launch the omnipresent SimpleHTTPServer. This was just an example application; any other application could be launched simply by changing the command line. We also learned how to programmatically query for information about the status of applications on Marathon.

Next, we learned about scaling applications on Marathon. Most importantly, we discussed placement constraints, and how to easily ensure that applications run on the correct class of slave machines and are spread evenly amongst racks.

Docker is the most popular application containerization technology. We learned how to launch Dockerized applications on Marathon with minimal fuss, and how to configure various Docker-specific options, such as mounting host disks.

Our applications wouldn’t be fully robust without including health checks and automatic incremental rollouts and rollbacks. We discussed how to configure TCP, HTTP, and command-based health checks, as well as how to specify the application’s behavior during upgrades. We also discussed how to subscribe to an event stream from Marathon to build custom integrations.

Then, we learned about integrating Marathon with a service discovery proxy. Bamboo, an open-source HAProxy configuration manager, is a powerful tool for synchronizing HAProxy with Marathon tasks. We also learned about popular approaches for building scalable, multilanguage microservices architectures with HAProxy and Marathon.

Finally, we looked at how to use Marathon to host other Mesos frameworks’ schedulers, using Chronos, a distributed cron-like framework, as an example. We also briefly surveyed alternative PaaS frameworks to Marathon, since they offer many additional features (at the cost of a steeper learning curve).

Perhaps Marathon and its alternatives don’t satisfy the needs of your application, and you need more control over how processes are launched or how failures are handled. In that case, you’ll need to write your own framework. In the next chapter, we’ll learn about the techniques used to build Marathon, so that you can build your own custom Mesos framework.

The Scheduler

The scheduler is the component that interacts directly with the leading Mesos master. A scheduler has four responsibilities:

1. Launch tasks on the received offers.

2. Handle status updates from those tasks, particularly to respond to task failures and crashes.

3. Persist state and manage failovers in order to be highly available.

4. Interact with clients, users, or other systems (since no system runs in a vacuum!).

Some of these responsibilities are implemented purely by interacting with the Mesos API; others must be implemented using the language of choice’s platform. For instance, launching tasks and handling status updates are implemented entirely by Mesos callbacks and API requests. On the other hand, it’s up to you to start a web server (such as Jetty on Java or Flask on Python) to allow clients to interact with the scheduler. High availability usually requires using both Mesos APIs and other libraries to create a solution that is suitable for the problem domain.

But how can we launch tasks if we’re only focusing on the scheduler? For our convenience, Mesos comes with something called the CommandExecutor, which is specifically designed to make it easy to launch and orchestrate command-line programs and Docker containers. Let’s take a look at some high-level designs for several common scheduler abstractions: building a pool of servers, a work queue, and a job processor.

First, we’ll look at the pool of servers design. Its architecture is illustrated in Figure 4-1.

Pool of Servers Scheduler

Figure 4-1. Architecture and information flow of a pool of servers scheduler

This scheduler is simply going to ensure that there are always X copies of some application, such as Ruby on Rails, running. To build a pool of servers, our scheduler will have a list of all the servers it wants to be running. Initially, the list will have all the servers in the pending state. As the scheduler receives offers, it will attempt to launch servers, putting them into the staging state. Once the servers start successfully, the scheduler can move them to the running state. If they ever crash, they’ll go back to the initial pending state so that they can be relaunched on a new offer. Note that we need three states here: pending, staging, and running.

You might be tempted to remove the staging state, since, after all, that seems awfully similar to pending—in both cases, the server is not yet started. But here’s the problem: Mesos is an asynchronous system! You could receive many offers between attempting to launch a task and having the task actually start. Thus, the staging state ensures that we only try to launch each server once, and only retry if the launch fails. If you forget to have a staging state, you might encounter a bug in which there are tens or hundreds of instances of a task when there should be only one.

Work Queue Scheduler

The work queue architecture, illustrated in Figure 4-2, allows clients to submit “work items” to a queue, where they’ll eventually be processed by workers. A work queue scheduler will ensure that we always have a sufficient number of workers running, so that there will be a sufficient number of workers to process the queued work items.

Figure 4-2. Architecture and information flow of a work queue scheduler

To build this scheduler, we’re actually going to cheat and assume that we already have a pool of servers scheduler, as described in the previous section. This way, we simply start the workers as servers in the pool. Therefore, the meat of this system will reside in the workers themselves.

Our workers will retrieve their work items from some known location. This location could be passed to them from the scheduler as a command-line option or environment variable. Each worker will connect to the queue, and then loop forever: retrieve an item, process it, and mark it as done.

The neat thing about considering work queues (and all potential frameworks) in this way is that you don’t even need to write a scheduler—you could just use Marathon or some other pool of services scheduler. Of course, this shifts the implementation of the queue to another system, such as RabbitMQ or Redis. This is a good thing: the Mesos messaging APIs are designed for controlling the applications on the cluster and are not as well suited to other domains, such as transactional logic or high-performance messaging. When developing frameworks for Mesos, you must use appropriate data storage backends—queues, databases, and document stores are all likely useful, depending on the framework you’re building.

Job Processor Scheduler

The final abstraction we’ll consider is the job processor scheduler (Figure 4-3).

Figure 4-3. Architecture and information flow of a job processor scheduler

A job processor will accept a list of jobs from clients and then run those jobs on a cluster; a job is just a command line to invoke, along with the CPU and memory resources it needs. This scheduler will require a slightly more complex state machine, since we’ll also need terminal states for jobs (we’ll call them successful and failed). We’ll also track how many retries have been attempted, so that we can give up on jobs with no hope for success.

This scheduler’s state machine will move jobs from pending to staging when it attempts to launch them on an offer, and from staging to running when the job actually starts running. Handling completion of a task is trickier, though: we’ll want to move the job to successful if it succeeds, to failed if it has exceeded the maximum number of retries, or to pending if it has more retries remaining. Furthermore, we could determine whether the task failed due to a bug in its code, a transient machine failure, or some other issue—this is useful feedback for deciding whether to have the job fail fast, or let it continue to retry.

Now that we’ve seen conceptually how a framework could be implemented with just a scheduler, let’s try our hand at implementing the job processor.

Useless Remote BASH

Before we get started on the job processor, let’s break down the absolute minimal code needed to get a task to launch on Mesos. We’ll call this program Useless Remote BASH,¹ as it’ll just run the hardcoded program in a container on the cluster.

Let’s break down this most simple of frameworks into several parts, and see why each part is necessary. In Example 4-1, we can see the imports and class declaration of our new scheduler.

package com.example;
import java.nio.file.*;
import java.util.*;
import org.apache.mesos.*;
import org.apache.mesos.Protos.*;

public class UselessRemoteBASH implements Scheduler {
    // We'll discuss the inner bits as we go
}

Two things here deserve our attention:

1. We import everything from org.apache.mesos and org.apache.mesos.Protos, so that we can use the Mesos APIs.

2. Schedulers should implement the Scheduler interface, which defines the messages a scheduler can receive as callbacks.

We’ll use the complementary SchedulerDriver class to send messages to Mesos. Example 4-2 is the entry point of our program.

public static void main(String ... args) throws Exception {
    FrameworkInfo frameworkInfo = FrameworkInfo.newBuilder()
        .setUser("")
        .setName("Useless Remote BASH")
        .build();

    Scheduler mySched = new UselessRemoteBASH();


    SchedulerDriver driver = new MesosSchedulerDriver(
        mySched,
        frameworkInfo,
        "zk://" + args[0] + "/mesos"
    );
    driver.start();
    driver.join();
}

In the main function, first, we construct the FrameworkInfo of our framework. FrameworkInfo is a protobuf that describes the framework to Mesos. You’ll want to choose a clear name for the framework, so that you can identify your framework in the Mesos UI. In practice, you’ll want to construct the name to include several pieces of data: which framework this is, as well as other high-level information about the framework, such as whether it’s a dev or prod instance. It’s entirely up to you what you name your framework, but consider the anecdote in the following sidebar!

Typically, we set the framework’s user to an empty string (""), which means that it will register with Mesos as whichever user is running the scheduler. In “Roles”, we discussed how the framework’s user can be used for quotas, capacity planning, and security. Most of the other fields of FrameworkInfo are used to configure high availability, so we’ll postpone discussing those until later.

After we’ve constructed our FrameworkInfo, we can create the driver, which is the component that actually communicates with Mesos. The driver for the scheduler is an instance of SchedulerDriver; you should stick with the MesosSchedulerDriver that is built with Mesos for now. In the future, other implementations may be written in order to remove the dependencies on native code. Unfortunately, the current projects working on this are more trouble than they’re worth for framework developers, as they haven’t reached stability yet.² The SchedulerDriver takes three arguments: an implementation of the Mesos Scheduler interface, the FrameworkInfo, and the address of the Mesos cluster. At this point, when we call start(), our driver will connect to the Mesos cluster, register the framework, and begin receiving offers. Usually, you’ll want to call join() at the end of your main function. join() blocks until the driver terminates, either from having stop() called on it or because Mesos instructed the driver to stop because the framework failed over.

Now that we’ve seen how frameworks start up, let’s look at Example 4-3. This function is an example of how to create a Mesos task that we can launch. As this is a useless framework, we hardcode what would normally be important parameters: the cpus, mem, and command we want to run.

public static TaskInfo makeTask(SlaveID targetSlave) {
    double cpus = 1.0;
    double mem = 100.0;
    String command = "echo hello world";

    UUID uuid = UUID.randomUUID();
    TaskID id = TaskID.newBuilder() 
        .setValue(uuid.toString())
        .build();
    return TaskInfo.newBuilder() 
        .setName("useless_remote_bash.task " + id.getValue())
        .setTaskId(id)
        .addResources(Resource.newBuilder() 
                .setName("cpus")
                .setType(Value.Type.SCALAR)
                .setScalar(Value.Scalar.newBuilder().setValue(cpus)))
        .addResources(Resource.newBuilder() 
                .setName("mem")
                .setType(Value.Type.SCALAR)
                .setScalar(Value.Scalar.newBuilder().setValue(mem)))
        .setCommand(CommandInfo.newBuilder().setValue(command)) 
        .setSlaveId(targetSlave) 
        .build();
}

We’ve managed to register our framework, and we’ve learned how to specify a task to launch. The last step is to actually implement that pesky Scheduler interface. For this super-basic framework, we’re going to ignore most of the callbacks, as they update us on information we don’t want to react to. Note that all of the callbacks provide you with a driver; this driver is guaranteed to be the same one you called start() on. Mesos passes you this as a convenience, since it’s the only way to interact with the cluster, and that’s typically something you’d like to do in response to cluster events. Example 4-4 shows the callbacks we want to log: registered and statusUpdate.

public void registered(SchedulerDriver driver, FrameworkID frameworkId,
                               MasterInfo masterInfo) {
    System.out.println("Registered with framework id " + frameworkId);
}

public void statusUpdate(SchedulerDriver driver, TaskStatus status) {
    System.out.println("Got status update "+status);
}

When we have successfully registered with the Mesos masters, we’ll get the registered callback. When you register a framework for the first time, you don’t specify the FrameworkID; instead, Mesos will assign it for you. This callback is the only place to learn the assignment of the FrameworkID. (In Example 4-15 and Example 4-14, we’ll see how we can use this FrameworkID to safely fail a scheduler over to a new instance, so that we can build zero-downtime systems.) The MasterInfo tells you which port and version the master is running on. Frameworks can decide which Mesos features are valid to use based on the master’s version.

The statusUpdate callback will soon become very important for our scheduler, as it’s how we track the lifecycle of our tasks: when they start running, and when and why they fail. Even though we’re not going to do anything with this callback yet, it’s convenient to log when it happens, as that will help us debug our framework and confirm tasks that are actually launching and finishing. Whenever I build a production framework, I always make sure to enable logging for every statusUpdate callback: those logs have helped me identify and debug issues countless times.

Finally, we get to look at the core of our new framework, as shown in Example 4-5. This callback is how we receive offers from Mesos—offers that are chock-full of juicy resources that we can launch tasks on.

private boolean submitted = false;

public void resourceOffers(SchedulerDriver driver, java.util.List<Offer> offers) {
    synchronized (this) {
        if (submitted) { 
            for (Offer o : offers) { 
                driver.declineOffer(o.getId());
            }
            return;
        }
        submitted = true;
        Offer offer = offers.get(0);
        TaskInfo ti = makeTask(offer.getSlaveId()); 
        driver.launchTasks( 
            Collections.singletonList(offer.getId()),
            Collections.singletonList(ti)
        );
        System.out.println("Launched offer: " + ti);
    }
}

In our framework, we first check if we’ve already submitted our task.

If we have, we simply decline all the offers.

If we haven’t submitted our task yet, we construct a TaskInfo for it with the ID of the slave from the first offer we received.

Then, we call launchTasks to attempt to launch the task!

The other Scheduler callbacks, listed in Example 4-6, can all be ignored for now.

public void disconnected(SchedulerDriver driver) { }
public void error(SchedulerDriver driver, java.lang.String message) { }
public void executorLost(SchedulerDriver driver, ExecutorID executorId,
                             SlaveID slaveId, int status) { }
public void frameworkMessage(SchedulerDriver driver, ExecutorID executorId,
                                 SlaveID slaveId, byte[] data) { }
public void offerRescinded(SchedulerDriver driver, OfferID offerId) { }
public void reregistered(SchedulerDriver driver, MasterInfo masterInfo) { }
public void slaveLost(SchedulerDriver driver, SlaveID slaveId) { }

And there you have it! You’ve written your first Mesos framework, and it actually does quite a lot. It can register with the master, create a TaskInfo to express how to launch a task, log status updates, and accept an offer to launch the task.

This Framework actually has two problematic bugs: when it accepts the offer to launch a task, it fails to check if the offer is actually big enough. If the offer was too small, we’ll get a TASK_LOST status update about the task, rather than anything indicating success. The other bug is that when it accepts the offer, it doesn’t decline the other offers that were potentially received in the resourceOffers callback. Those other offers will be stuck in a limbo state until they expire, which could be several minutes (or forever, if you didn’t change the default). They’ll be unavailable to other frameworks in the meantime.

In the coming sections, we’ll improve resourceOffers to fix these issues.

Implementing a Basic Job Processor

Now, let’s extend our framework to be more useful by reading a list of jobs from a JSON file, and then having it launch all of those jobs on Mesos. We’ll start by looking at how we’ll model a job (Example 4-7).

public class Job {
    private double cpus; 
    private double mem; 
    private String command; 
    private boolean submitted; 

    private Job() {
        submitted = false;
    }

    public TaskInfo makeTask(SlaveID targetSlave) { 
        UUID uuid = UUID.randomUUID();
        TaskID id = TaskID.newBuilder()
            .setValue(uuid.toString())
            .build();
        return TaskInfo.newBuilder()
            .setName("task " + id.getValue())
            .setTaskId(id)
            .addResources(Resource.newBuilder()
                    .setName("cpus")
                    .setType(Value.Type.SCALAR)
                    .setScalar(Value.Scalar.newBuilder().setValue(cpus)))
            .addResources(Resource.newBuilder()
                    .setName("mem")
                    .setType(Value.Type.SCALAR)
                    .setScalar(Value.Scalar.newBuilder().setValue(mem)))
            .setSlaveId(targetSlave)
            .setCommand(CommandInfo.newBuilder().setValue(command))
            .build();
    }

    public static Job fromJSON(JSONObject obj) throws JSONException { 
        Job job = new Job();
        job.cpus = obj.getDouble("cpus");
        job.mem = obj.getDouble("mem");
        job.command = obj.getString("command");
        return job;
    }

    // ... snip ... 
}

The Job is a container for information about a particular job, like a model class in MVC. We’ve only added two special methods: fromJSON, which gives us a way to construct the Job with valid starting state, and makeTask, which gives us a way to easily turn it into a TaskInfo without having all the protobuf munging code in our scheduler class.

Next, let’s look at how our main function has evolved. Example 4-8 shows the revised version.

public static void main(String ... args) throws Exception {
    byte[] data = Files.readAllBytes(Paths.get(args[1])); 
    JSONObject config = new JSONObject(new String(data, "UTF-8")); 
    JSONArray jobsArray = config.getJSONArray("jobs");
    List<Job> jobs = new ArrayList<>();
    for (int i = 0; i < jobsArray.length(); i++) { 
        jobs.add(Job.fromJSON(jobsArray.getJSONObject(i)));
    }

    System.out.println(jobs);

    // Below here, the function remains almost as before
    FrameworkInfo frameworkInfo = FrameworkInfo.newBuilder()
        .setUser("")
        .setName("Useless Remote BASH")
        .build();

    Scheduler mySched = new UselessRemoteBASH(jobs); 
    SchedulerDriver driver = new MesosSchedulerDriver(
        mySched,
        frameworkInfo,
        "zk://" + args[0] + "/mesos"
    );
    driver.start();
    driver.join();
}

This allows us to read the entire contents of a file into memory. We assume that the first argument is still the Mesos cluster, and the second argument is the name of a file that contains the JSON configuration.

We convert the byte array to JSON in this step, so that we can extract the Job configurations.

Finally, we loop over the JSON job descriptors, processing them into Jobs.

This time, we pass an argument to our Scheduler: the jobs we want it to launch.

Let’s take a look at a sample JSON file of jobs, to get a feel for the schema:

{
    "jobs": [ 
        {
            "cpus": 0.5, 
            "mem": 100,
            "command": "sleep 60; echo hello world"
        }
    ]
}

The JSON file should contain a JSON object with a single key, jobs. That key’s value should be a list of Job objects.

Each job object has three properties: cpus, mem, and command, which correspond to the eponymous fields in the Job class.

Now that we can load jobs up, let’s see how we evolved resourceOffers. The new version is shown in Example 4-9.

public void resourceOffers(SchedulerDriver driver, java.util.List<Offer> offers) {
    synchronized (jobs) {
        List<Job> pendingJobs = new ArrayList<>(); 
        for (Job j : jobs) { 
            if (!j.isSubmitted()) {
                pendingJobs.add(j);
            }
        }
        for (Offer o : offers) { 
            if (pendingJobs.isEmpty()) { 
                driver.declineOffer(o.getId());
                break;
            }
            Job j = pendingJobs.remove(0); 
            TaskInfo ti = j.makeTask(o.getSlaveId());
            driver.launchTasks(
                Collections.singletonList(o.getId()),
                Collections.singletonList(ti)
            );
            j.setSubmitted(true); 
        }
    }
}

Because our Jobs contain state information (i.e., whether or not they were just launched), we need to first calculate which jobs haven’t yet been launched.

This loop allows us to find all the jobs that haven’t been submitted yet, and add them to pendingJobs.

Now, we’re going to try to match jobs to offers.

If we have no more jobs we want to launch, then we can decline the offer.

If we reach this line, we have a job to launch. From here on, the code is similar to our first incarnation.

Remember that we want to mark the job as submitted, or else we could accidentally double-submit the job.

This version of resourceOffers shows us several useful patterns. Every framework needs to keep track of the running tasks and work that it wants to do in the future. For most frameworks, it will be most convenient to have a single data structure that tracks all ongoing work. This data structure will include both running and pending tasks. Of course, when you actually want to launch a task, you’ll need to calculate the specific set of tasks that are still pending. In this example, after each resource offer, we compute the set of pending tasks. Later in this chapter, we’ll look at other architectures for tracking this information.

The other pattern that every framework uses is looping over all the offers, matching offers to pending work until one or the other is exhausted. But you might be thinking: resourceOffers requires a lot of care, and this framework still fails to handle it effectively! You’d be absolutely correct, which is why we’re going to look at how to fix that problem now.

Matching Tasks to Offers

As we’ve been alluding to throughout this chapter, the assignment of tasks to offers is fraught with details that must be attended to. For instance, we must ensure that the offer has all the necessary resources for our job, we must attempt to launch as many tasks as we can on each offer, and we must prioritize which task we’ll launch next. The reason we want to match as many tasks as possible to each offer is that if we only match one task per offer, our scheduler will launch tasks at a slow rate. Specifically:

With the default offer interval of 1 second, if you have a five-slave cluster, you’ll only be able to launch at most 60 tasks per minute. With properly implemented offer handling code, however, you could launch tens of thousands of tasks per minute! For our example, we’re going to implement the first fit algorithm, since it requires very little code. Then, we’ll learn about other algorithms, and various considerations therein. Let’s make the following change to our resourceOffers:

... // code unchanged
driver.launchTasks(
    Collections.singletonList(o.getId()),
    doFirstFit(o, pendingJobs);
);
... // code unchanged

Instead of directly taking a job, we’re going to compute the first fit of all the pending jobs to the given offer. The solution we find is a List of tasks, since the whole point is that we can launch multiple tasks on a single offer.

Obviously, the interesting part is the implementation of doFirstFit (Example 4-10). The idea of first fit is this: given the amount of space available in an offer, we’ll add tasks that fit in whatever space is left in the offer. When we decide to add a task, we’ll deduct its space usage from the remaining space. Once we can’t fit any more tasks into the offer, we’ll remove all the matched Jobs from the list of Job provided as an argument, and we’ll return the TaskInfos that will launch those Jobs.

public List<TaskInfo> doFirstFit(Offer offer, List<Job> jobs) {
    List<TaskInfo> toLaunch = new ArrayList<>();
    List<Job> launchedJobs = new ArrayList<>();
    double offerCpus = 0; 
    double offerMem = 0;
    // We always need to extract the resource info from the offer.
    // It's a bit annoying in every language.
    for (Resource r : offer.getResourcesList()) { 
        if (r.getName().equals("cpus")) {
            offerCpus += r.getScalar().getValue();
        } else if (r.getName().equals("mem")) {
            offerMem += r.getScalar().getValue();
        }
    }
    // Now, we will pack jobs into the offer
    for (Job j : jobs) {
        double jobCpus = j.getCpus();
        double jobMem = j.getMem();
        if (jobCpus <= offerCpus && jobMem <= offerMem) { 
            offerCpus -= jobCpus;
            offerMem -= jobMem;
            toLaunch.add(j.makeTask(offer.getSlaveId()));
            j.setSubmitted(true);
            launchedJobs.add(j); 
        }
    }
    for (Job j : launchedJobs) {
        j.launch(); 
    }
    jobs.removeAll(launchedJobs);
    return toLaunch;
}

We’ll keep track of how many resources are remaining in the offer in these variables.

You must iterate over all the resources in the offer to find the ones whose names match the type of resource you care about, and only then can you extract the value. The reason this cannot be a map is that you can receive multiple resources of the same name; for example, you could receive both reserved and unreserved memory resources from a slave.

Here, we check to see if the offer still has enough resources remaining to launch our task. If it does, we’ll deduct those resources from our method’s internal count, add the TaskInfo to the list of tasks we will launch, and mark the job as submitted.

In order to support doFirstFit semantics, we add all jobs we’ll launch to a list, so that we can remove all those elements from the original list of all pending jobs.

Job has methods that help us track its instances’ progress through their state machine. We’ll go over this in Example 4-11.

First fit is usually the best algorithm to use when matching tasks to offers. You might think that this won’t always utilize offers as efficiently as if you put more work into trying to optimize matching the offers. That’s absolutely true, but consider the following: a cluster either has sufficient or insufficient resources for launching every pending task. If there are plenty of resources, then first fit should always be able to launch every task. If, on the other hand, there are not enough resources, nothing would be able to launch every task. Thus, it’s a good idea to write code to choose which task to run next, so that you can maintain the target quality of service. Only if there are just barely enough resources does more sophisticated packing begin to make sense. Unfortunately for us, this problem—known more generally as the knapsack problem—is a notorious NP-complete problem. NP-complete problems are those problems for which it can be proved that an optimal solution will take an extremely long time to find, and there’s not a clever trick in existence to solve the problem faster.³

Actually, the situation isn’t so bad. In fact, although it’s impossible to solve the packing problem perfectly and quickly, there are several techniques that give “good enough” results and aren’t too tricky to implement. For some frameworks, all tasks will be bottlenecked on a single resource: either CPUs (for compute-bound workloads, like data analysis) or memory (i.e., caches like memcached or Redis). As a result, when we try to pack our resources, we can simplify the packing problem to a single dimension: the most constrained dimension. Once we’ve done this, we can use techniques such as the fully polynomial time greedy approximation scheme for the 0/1 knapsack problem.⁴

Bridging the Semantic Gap Between Offers and Jobs

Now, we’ve got a pretty cool framework. It takes a JSON configuration file filled with the jobs we want to run, and it submits those jobs to the Mesos cluster efficiently. We have a problem, though—what if a job fails? At the moment, we’ll just see a bit of output in the terminal indicating that the task failed, succeeded, or was lost. We want to add support for retries: when a job fails, we’ll resubmit it in the hope that it will succeed this time. Eventually, however, we’ll give up and consider it truly failed. To this end, we’re going to expand our Job class so that instead of simply having a Boolean property submitted, it will have a JobState:

public enum JobState {
    PENDING,
    STAGING,
    RUNNING,
    SUCCESSFUL,
    FAILED
}

We need to increase the number of states a job can be in in order to support retries. Before, we would simply launch unsubmitted jobs. Now, we’ll need to track the exact part of the task lifecycle a job is in, and resubmit it only if it has already failed. Figure 4-4 illustrates the valid state transitions.

Figure 4-4. Valid transitions of JobState

We’ll also have to expand the Job class to internally keep track of retries and the job’s state. Example 4-11 shows the new methods that will implement the preceding state transition diagram.

Example 4-11. State transition methods for Job

public class Job {
    // previous fields elided
    private int retries;

    private Job() {
        status = JobState.PENDING;
        id = UUID.randomUUID().toString();
        retries = 3;
    }

    public void launch() {
        status = JobState.STAGING;
    }

    public void started() {
        status = JobState.RUNNING;
    }

    public void succeed() {
        status = JobState.SUCCESSFUL;
    }

    public void fail() {
        if (retries == 0) {
            status = JobState.FAILED;
        } else {
            retries--;
            status = JobState.PENDING;
        }
    }
}

We’ve decided to give every job three retries. To simplify the implementation of the state changes of the job, we have four functions to move between states: launch(), started(), succeed(), and fail(). These functions will be called, respectively, when we submit the task to Mesos, when it starts, and when it succeeds or fails. Example 4-12 shows how we move between states.

Example 4-12. Enhanced statusUpdate handler

public void statusUpdate(SchedulerDriver driver, TaskStatus status) {
    synchronized (jobs) {
        // We'll see if we can find a job this corresponds to
        for (Job j : jobs) { 
            if (j.getId().equals(status.getTaskId().getValue())) { 
                switch (status.getState()) {
                    case TASK_RUNNING: 
                        j.started();
                        break;
                    case TASK_FINISHED: 
                        j.succeed();
                        break;
                    case TASK_FAILED: 
                    case TASK_KILLED: 
                    case TASK_LOST: 
                    case TASK_ERROR: 
                        j.fail();
                        break;
                    default:
                        break;
                }
            }
        }
    }
}

Typically, one should store a mapping between the TaskID and the Job, rather than doing a linear search.

This task status, running, means that we’ve received confirmation that the task was successfully started by the slave.

A finished task is one that completed without reporting any kind of error.

Failed tasks are those that did not complete successfully, and whose termination was initiated through a normal codepath by the task itself. For example, if a task that is supposed to download a file wasn’t able to do so, it should report its status as failed.

Killed tasks are those that did not complete successfully because the scheduler told them to stop. When a scheduler performs preemption or allows a user to request the early termination of a task, that task should report the status killed.

When a task is lost, that means that an unexpected error occurred. That could be due to the slave unexpectedly going offline, or an executor that exited before it sent a TASK_FINISHED, TASK_KILLED, or TASK_KILLED message.

This is a very rare case, indicating a problem with the task descriptor itself. You can look in the TaskStatus’s message field for details.

Adding High Availability

At this point, our scheduler-only framework is nearly feature-complete. We are loading jobs from configuration files, launching them on Mesos, and automatically restarting them when they fail. Now it’s time to add the final feature: high availability.

High availability means that when the scheduler process fails, the framework as a whole continues to run. This is accomplished via the cooperation of three subfeatures:

1. The scheduler will use leader election so that we can run multiple instances and have a fallback instance automatically take over when the current leader fails.

2. The scheduler must synchronize its state to shared, distributed storage, so that the newly elected leader can pick up where the last one left off.

3. The scheduler must opt into a few Mesos features that will ensure that running tasks are unaffected by scheduler failure.

For this example, we’re going to use ZooKeeper to provide our leader election and distributed storage, since Mesos clusters usually⁵ come with a ZooKeeper cluster. Furthermore, we’re going to use the Apache Curator library to interact with ZooKeeper, since it handles many of the challenges of the included ZooKeeper API.⁶ Since this isn’t meant to be a production-quality framework, we’ll hardcode all our framework’s data to reside under the /sampleframework path in ZooKeeper. For production-quality frameworks, you should make the location of all data configurable in ZooKeeper. Example 4-13 shows how we add basic leader election to our main function.

public static void main(String ... args) throws Exception {
    CuratorFramework curator = CuratorFrameworkFactory.newClient( 
            args[0],
            new RetryOneTime(1000)
    );
    curator.start(); 

    LeaderLatch leaderLatch = new LeaderLatch(curator, "/sampleframework/leader"); 
    leaderLatch.start(); 
    leaderLatch.await(); 

    // The previous main implementation goes here
}

Here, we see how to create a new instance of Curator. It takes the address of the ZooKeeper cluster and a RetryPolicy as arguments. While we are using the RetryOneTime policy in this book, this is probably not the right choice for production frameworks. Instead, the ExponentialBackoffRetry or BoundedExponentialBackoffRetry policies are more robust choices.

Note that you must explicitly start() the Curator framework.⁷

Next, we create a LeaderLatch for our framework, which we’ll put under the ZooKeeper path we chose for our framework, /sampleframework.

Never forget to start() Curator components—this is the #1 pitfall.

The await() method on a started LeaderLatch will not return until we are the leader.

Unfortunately, it’s not enough to simply allow multiple instances of our framework to elect a leader from amongst themselves. Our goal is to have the new leader actually become the same framework as before. To do this, we’ll need to inform Mesos that our scheduler supports failover. In addition, when the new leader registers, it needs to tell Mesos that it’s the new scheduler for our framework. In Mesos, schedulers are identified by their FrameworkID, an optional value in the FrameworkInfo. The FrameworkID must be assigned by Mesos in order to ensure it remains unique for each framework. Now, we’ll need to actually store the FrameworkID when it’s assigned, so that future leaders can reuse it.

We’re going to store the FrameworkID in the ZooKeeper node id under our /sampleframework path. When we start up, we’ll first check to see if there’s already a saved FrameworkID; if so, we’ll start up using that. Otherwise, we’ll operate as before, and allow the Mesos master to assign a FrameworkID (see Example 4-14). In addition, we will store whatever FrameworkID Mesos informs us we registered with, so that future leaders can reuse that ID. We can store that FrameworkID unconditionally, since it will never change after the first assignment, and thus is idempotent.⁸

FrameworkInfo.Builder frameworkInfoBuilder = FrameworkInfo.newBuilder()
    .setUser("")
    .setName("Useless Remote BASH"); 

try {
    byte[] curatorData = curator.getData().forPath("/sampleframework/id"); 
    frameworkInfoBuilder.setId(new String(curatorData, "UTF-8"));
} catch (KeeperException.NoNodeException e) {
    // Don't set the FrameworkID 
}

FrameworkInfo frameworkInfo = frameworkInfoBuilder
    .setFailoverTimeout(60*60*24*7) 
    .build();

We start with our builder, the same as usual.

We attempt to fetch the stored FrameworkID. If this node doesn’t exist in ZooKeeper, we throw a KeeperException.NoNodeException.

If there wasn’t a stored FrameworkID, we allow Mesos to assign it.

We must configure the framework to have a failover timeout, in seconds. In this case, we allow failover to happen for up to a week. This means we’ve got a week to reconnect a new scheduler for this framework after shutting down the previous scheduler. This timeout should be set to help you survive a worst-case scenario when you need enough time to recover the system after a catastrophe. After this timeout, requests to register with the old FrameworkID will be met with an error about registering a finished framework, and all tasks that were running under that framework will be killed.

Now, we need to update our registered scheduler callback to store the FrameworkID. The new version is shown in Example 4-15.

public void registered(SchedulerDriver driver, FrameworkID frameworkId,
                                 MasterInfo masterInfo) {
    System.out.println("Registered with framework id " + frameworkId);
    try {
        curator.create().creatingParentsIfNeeded().forPath( 
            "/sampleframework/id",
            frameworkId.getBytes()
        );
    } catch (KeeperException.NodeExistsException e) {
        // Do nothing 
    }
}

We attempt to store the data of the FrameworkID to the predetermined location.

If we catch this exception, it means that we’ve already created the node /sampleframework/id. In this case, we’ll assume that means an earlier leader scheduler already stored the ID, so we’ll do nothing. If you’re using a different store than ZooKeeper, it may be more convenient to simply unconditionally store the ID every time, since it will never change.

Now, we’ve finally got a highly available framework with failover working. Or at least, we would if the framework had any way to also synchronize its state with regard to which jobs are running. To do this, we’ll need to add support for serializing and deserializing jobs from ZooKeeper. We’ll store each job in ZooKeeper at /sampleframework/jobs/$jobid, where $jobid is the id field of the job. Serialization can be trivially implemented via JSON.

Now that our framework is highly available, we don’t want to always have to provide the set of jobs it should run. In fact, we’d like to have our framework automatically pick up the monitoring of all previously provided jobs on the command line, and also optionally launch new jobs provided on the command line. To do this, we’ll load jobs both from the command line (if provided) and from ZooKeeper, as shown in Example 4-16.

List<Job> jobs = new ArrayList<>(); 

if (args.length > 1) { 
    byte[] data = Files.readAllBytes(Paths.get(args[1]));
    JSONObject config = new JSONObject(new String(data, "UTF-8"));
    JSONArray jobsArray = config.getJSONArray("jobs");
    for (int i = 0; i < jobsArray.length(); i++) {
        jobs.add(Job.fromJSON(jobsArray.getJSONObject(i), curator));
    }
    System.out.println("Loaded jobs from file");
}

//Load jobs from ZK
try {
    for (String id : curator.getChildren().forPath("/sampleframework/jobs")) { 
        byte[] data = curator.getData()
                             .forPath("/sampleframework/jobs/" + id); 
        JSONObject jobJSON = new JSONObject(new String(data, "UTF-8"));
        Job job = Job.fromJSON(jobJSON, curator); 
        jobs.add(job);
    }
    System.out.println("Loaded jobs from ZK");
} catch (Exception e) {
    // Sample code isn't production ready 
}

We start by creating the List we’ll store all the Jobs in.

We only run the original job loading code if a filename was passed on the command line.

The getChildren() method in Curator allows us to get the names of all the child nodes of a path. These child nodes are saved after most state-changing methods, such as launch(), started(), and succeed().

For each job, we’ll fetch the serialized JSON representation from ZooKeeper.

At this point, we can use the original JSON processing code to deserialize the Job. Note that we must provide every Job with a reference to the Curator now, since the Jobs must serialize themselves.

In real production code, you should probably handle exceptions.

We’ll also need to enhance our Jobs to be able to save their state, as shown in Example 4-17.

public class Job {
    // Previously defined fields elided
    private String id; 

    private void saveState() { 
        JSONObject obj = new JSONObject();
        obj.put("id", id); 
        obj.put("status", (status == JobState.STAGING ?
            JobState.RUNNING : status).toString()); 
        // Storing other fields omitted for brevity
        byte[] data = obj.toString().getBytes("UTF-8");
        try { 
            curator.setData().forPath("/sampleframework/jobs/" + id, data);
        } catch (KeeperException.NoNodeException e) {
            curator.create()
                   .creatingParentsIfNeeded()
                   .forPath("/sampleframework/jobs/" + id, data);
        }
    }







    public void launch() {
        // ... previous code omitted ...
        saveState(); 
    }

    // started(), succeed(), fail() are modified as above
}

We need to include an identifier for each job, so that they can be looked up by a primary key in the database we sync them to.

We’ve added a method to save the state of the Job to an external store. In this case, we’re storing the JSON-formatted data to ZooKeeper.

We need to make sure to store all the fields to the database.

We always store jobs that are in STAGING as if they were running. Since we can’t be sure whether a job in the STAGING state has launched successfully, we’ll assume it succeeded during a failover, so that we’ll either get a new statusUpdate if it finished in the meantime or a TASK_LOST after reconciliation.

Curator’s API requires a different method to create a new ZooKeeper node and update its data.

We need to make sure to save the Job’s state after every status update. Note that the addition of saveState() to the last line of started(), succeed(), and fail() is omitted in this code, but this must be done for a working framework.

And there you have it! Our job scheduler is now able to automatically fail over to waiting hot spares without impacting the running jobs or losing track of them—well, almost.

Adding Reconciliation

It turns out that there are still situations in which things can fall out of synchronization: these cases are explained in “Reconciliation During Failover”. This is the final piece missing from our robust, highly available Mesos scheduler. Luckily, it’s easy to add reconciliation to a scheduler. When implementing the statusUpdate() callback, you are mapping the Mesos state events to your framework’s state machine model. As long as this state machine never moves out of the terminal states (e.g., TASK_LOST, TASK_FINISHED, TASK_KILLED, etc.), reconciliation shouldn’t require additional work. Typically, you’ll want to run reconciliation periodically; I usually just run it on startup and every 30 minutes thereafter. Reconciliation is a sort of catchall for any message loss—it ensures that even if some unknown bug or hardware failure causes your framework to fall out of sync with the state of the Mesos cluster, you’ll eventually correct that issue.

To reconcile, we will just generate a list of running tasks, along with the ID of the slave that they’re running on (for an explanation of why, see “Reconciliation During Failover”). Then, we’ll invoke reconcileTasks(). Mesos will automatically send the responses to the statusUpdate() callback. If you want to implement special handling of reconciliation-related status updates, you can check if the reason field of the status update is TASK_RECONCILIATION. We don’t bother with that in our framework, because our framework’s task state machine is robust. Example 4-18 shows how to reconcile the running tasks.

List<TaskStatus> runningTasks = new ArrayList<>();
for (Job job : jobs) {
    if (job.getStatus() == JobState.RUNNING) {
        TaskID id = TaskID.newBuilder().setValue(job.getId()).build();
        SlaveID slaveId = SlaveID.newBuilder().setValue(job.getSlaveId()).build(); 
        System.out.println("Reconciling the task " + job.getId());
        runningTasks.add(TaskStatus.newBuilder()
                .setSlaveId(slaveId)
                .setTaskId(id)
                .setState(TaskState.TASK_RUNNING)
                .build());
    }
}
driver.reconcileTasks(runningTasks);